package com.andnnl;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;

/**
 * Created by chenss on 2019/6/14.
 */
public class MoCount {
    public static void main(String[] args) {
        //[06-14 11:00:15,200 INFO ] [pool-1-thread-152] SMPPServerSimulator.run(1503) : Sending delivery mo for message id 4c39e9d8 content:

        SparkConf conf = new SparkConf();
        JavaSparkContext sc = new JavaSparkContext("local","word count local",conf);//本地跑
        JavaRDD<String> lines = sc.textFile("D:\\stdout.log.2019-06-11");
//        JavaRDD<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator()).filter(word->!word.isEmpty());
        JavaPairRDD<String, Integer> counts = lines
//                .filter(a->a.indexOf("Sending delivery mo for message")!=-1)
//                .filter(a->a.indexOf("Sending delivery reciept for messageId")!=-1)
                .filter(a->a.indexOf("returnMoMsgId")!=-1)
                .map(a->a.substring(7,15))
                .mapToPair(w -> new Tuple2<String, Integer>(w, 1))
                .reduceByKey((x, y) -> x + y)
                .sortByKey(true);
//        counts.collect().forEach(tuple->System.out.println(tuple._1 + " -> " + tuple._2));
        StringBuffer sb1=new StringBuffer();
        StringBuffer sb2=new StringBuffer();
        counts.collect().forEach(tuple->{
            sb1.append("'").append(tuple._1).append("',");
            sb2.append(tuple._2).append(",");
        });
        System.out.println(sb1.toString());
        System.out.println(sb2.toString());
//        JavaPairRDD<String, Integer> counts = words.mapToPair(w -> new Tuple2<String, Integer>(w, 1))
//                .reduceByKey((x, y) -> x + y)
//                .mapToPair(s -> new Tuple2<Integer, String>(s._2, s._1))
//                .sortByKey(false)
//                .mapToPair(s -> new Tuple2<String, Integer>(s._2, s._1));
//
//        counts.take(10).forEach(tuple->System.out.println(tuple._1() + " -> " + tuple._2()));

        sc.stop();
    }
}
